package cn.sciento.transfer.infra.rocketmq.vo;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import cn.sciento.core.base.BaseConstants;
import cn.sciento.transfer.infra.datasource.sql.SqlBuilder;

public class MqMessageSqlMapping {
  private Map<String, Object> rowData;
  
  private SqlInfo<?, ?> sqlMapping;
  
  private List<SqlInfo<?, ?>> tlSqlInfoList;
  
  private MqMessageSqlMapping() {}
  
  public SqlInfo<?, ?> getSqlMapping() {
    return this.sqlMapping;
  }
  
  public void setSqlMapping(SqlInfo<?, ?> sqlMapping) {
    this.sqlMapping = sqlMapping;
  }
  
  public static Builder builder() {
    return new Builder();
  }
  
  public List<SqlInfo<?, ?>> getTlSqlInfoList() {
    return this.tlSqlInfoList;
  }
  
  public void setTlSqlInfoList(List<SqlInfo<?, ?>> tlSqlInfoList) {
    this.tlSqlInfoList = tlSqlInfoList;
  }
  
  public Map<String, Object> getRowData() {
    return this.rowData;
  }
  
  public void setRowData(Map<String, Object> rowData) {
    this.rowData = rowData;
  }
  
  public static class Builder {
    private static final String TLS_FLAG = "tls";
    
    private Builder() {}
    
    public List<MqMessageSqlMapping> builderSqlInfo(MqMessage mqMessage) {
      if (StringUtils.isEmpty(mqMessage.getAction()) && 
        StringUtils.isEmpty(mqMessage.getType()))
        return null; 
      List<MqMessageSqlMapping> mqMessageSqlMappings = new ArrayList<>();
      List<Map<String, Object>> rows = mqMessage.getData();
      for (Map<String, Object> row : rows) {
        MqMessageSqlMapping mqMessageSqlMapping = new MqMessageSqlMapping();
        if (BaseConstants.Flag.YES.equals(mqMessage.getTlFlag())) {
          List<Map<String, Object>> tlRows = acquireTlRows(row, mqMessage);
          if (CollectionUtils.isNotEmpty(tlRows)) {
            List<SqlInfo<?, ?>> sqlInfoList = new ArrayList<>();
            for (Map<String, Object> map : tlRows) {
              SqlInfo<String, List<Object>> sqlInfo = builderSqlInfo(map, mqMessage, true);
              sqlInfoList.add(sqlInfo);
            } 
            mqMessageSqlMapping.setTlSqlInfoList(sqlInfoList);
          } 
        } 
        Object multiLangData = row.get("tls");
        try {
          row.remove("tls");
          mqMessageSqlMapping.setRowData(row);
          mqMessageSqlMapping.setSqlMapping(builderSqlInfo(row, mqMessage, false));
          mqMessageSqlMappings.add(mqMessageSqlMapping);
        } finally {
          row.put("tls", multiLangData);
        } 
      } 
      return mqMessageSqlMappings;
    }
    
    private SqlInfo<String, List<Object>> builderSqlInfo(Map<String, Object> row, MqMessage mqMessage, boolean multiFlag) {
      String sql;
      List<Object> params;
      SqlInfo<String, List<Object>> sqlInfo = new SqlInfo<>();
      sqlInfo.setAction(mqMessage.getAction());
      sqlInfo.setType(mqMessage.getType());
      switch (mqMessage.getAction()) {
        case "DELETE":
          sql = SqlBuilder.buildDelete(Collections.singletonList(mqMessage.getKeyField()), mqMessage
              .getTable());
          sqlInfo.setSql(sql);
          sqlInfo.setData(Collections.singletonList(row.get(mqMessage.getKeyField())));
          break;
        case "INSERT":
          sql = SqlBuilder.buildInsert(new ArrayList(row.keySet()), mqMessage.getTable());
          sqlInfo.setSql(sql);
          sqlInfo.setData(sortMapValueByKey(row));
          break;
        case "UPDATE":
          if (multiFlag) {
            sql = SqlBuilder.buildUpdate(new ArrayList(row.keySet()), 
                Arrays.asList(mqMessage.getKeyField(), "lang"), mqMessage
                .getTable());
          } else {
            sql = SqlBuilder.buildUpdate(new ArrayList(row.keySet()), 
                Collections.singletonList(mqMessage.getKeyField()), mqMessage
                .getTable());
          } 
          sqlInfo.setSql(sql);
          params = sortMapValueByKey(row);
          params.add(row.get(mqMessage.getKeyField()));
          if (multiFlag)
            params.add(row.get("lang")); 
          sqlInfo.setData(params);
          break;
      } 
      return sqlInfo;
    }
    
    private List<Object> sortMapValueByKey(Map<String, Object> map) {
      List<Object> values = new ArrayList();
      List<String> keySet = new ArrayList<>(map.keySet());
      for (String key : keySet)
        values.add(map.get(key)); 
      return values;
    }
    
    private List<Map<String, Object>> acquireTlRows(Map<String, Object> row, MqMessage mqMessage) {
      List<Map<String, Object>> tlRows = new ArrayList<>();
      Map<String, Object> tls = (Map<String, Object>)row.get("tls");
      for (Map.Entry<String, Object> columnEntry : tls.entrySet()) {
        Map<String, Object> langMap = (Map<String, Object>)columnEntry.getValue();
        for (Map.Entry<String, Object> columnLangEntry : langMap.entrySet()) {
          Map<String, Object> tlRow = new HashMap<>();
          tlRow.put("lang", columnLangEntry.getKey());
          tlRow.put(columnEntry.getKey(), columnLangEntry.getValue());
          tlRow.put(mqMessage.getKeyField(), row.get(mqMessage.getKeyField()));
          tlRows.add(tlRow);
        } 
      } 
      return tlRows;
    }
  }
}
